package com.ls.mq.pubsub;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.lang.ArrayUtils;

import com.ls.mq.iterface.MessageListener;
import com.ls.mq.iterface.MessageService;
import com.ls.mq.redis.RedisFactory;
import com.ls.mq.thread.ExecutorsManager;
import com.ls.mq.util.Toolkit;

import redis.clients.jedis.JedisPubSub;

public class PubSubProxy implements MessageService {
	/**
	 * channel 和监听者列表
	 */
	private Map<String, Set<MessageListener>> listenerMap;
	private JedisPubSub pubsubClient;
	private Set<String> subscribedChannels;
	private boolean isSubscribe;
	private volatile static PubSubProxy instance;

	public static PubSubProxy getInstance() {
		if (instance == null)
			synchronized (PubSubProxy.class) {
				if (instance == null)
					instance = new PubSubProxy();
			}
		return instance;
	}

	private PubSubProxy() {
		listenerMap = new HashMap<>();
		pubsubClient = new JedisPubSubListener(listenerMap);
		subscribedChannels = new HashSet<>();
	}

	@Override
	public void publish(String channel, String msg) {
		RedisFactory.getInstance().getClient().publish(channel, msg);
	}

	/**
	 * 可以向channel中添加多个监听者
	 * 
	 * @param channel
	 * @param listeners
	 *            类的全名
	 */
	@SuppressWarnings("unchecked")
	public synchronized void addListener(String channel, String... listenerNames) {
		MessageListener[] listeners = new MessageListener[listenerNames.length];
		int i = 0;
		for (String clazz : listenerNames) {
			try {
				Class<? extends MessageListener> cls = (Class<? extends MessageListener>) Class.forName(clazz);
				listeners[i++] = cls.newInstance();
			} catch (ClassNotFoundException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (InstantiationException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (IllegalAccessException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}

		}
		addListener(channel, listeners);
	}

	/**
	 * 可以向一个channel中添加多个listener，使得多个监听者可以同时收到同一个channel的信息
	 * 
	 * @param channel
	 * @param listeners
	 */
	public synchronized void addListener(String channel, MessageListener... listeners) {
		subcribeClient(channel);
		Set<MessageListener> existListener = listenerMap.get(channel);
		if (Toolkit.isEmpty(existListener)) {
			existListener = new HashSet<MessageListener>();
			listenerMap.put(channel, existListener);
		}
		for (MessageListener l : listeners) {
			existListener.add(l);
		}
	}

	/**
	 * 一个监听者注册到多个channel中
	 * 
	 * @param listeners
	 * @param channels
	 */
	public synchronized void addListener(MessageListener listener, String... channels) {
		subcribeClient(channels);
		for (String channel : channels) {
			Set<MessageListener> existListener = listenerMap.get(channel);
			if (Toolkit.isEmpty(existListener)) {
				existListener = new HashSet<MessageListener>();
				listenerMap.put(channel, existListener);
			}
			existListener.add(listener);
		}
	}

	private void subcribeClient(String... channels) {
		List<String> newChannels = new ArrayList<>();
		for (String channel : channels) {
			if (!subscribedChannels.contains(channel)) {
				subscribedChannels.add(channel);
				newChannels.add(channel);
			}
		}
		if (Toolkit.notEmpty(newChannels))
			subscribe(newChannels.toArray(new String[0]));
	}

	private void subscribe(String... channels) {
		if (isSubscribe)
			pubsubClient.subscribe(channels);
		else {
			subscribe(pubsubClient, channels);
			isSubscribe = true;
		}
	}

	private void subscribe(final JedisPubSub pubsubClient, final String[] channels) {
		ExecutorsManager.getInstance().submit(new Runnable() {
			@Override
			public void run() {
				RedisFactory.getInstance().getClient().subscribe(pubsubClient, channels);
			}
		});
		
	}

	private void unsubscribe(String... channels) {
		if (isSubscribe)
			pubsubClient.unsubscribe(channels);
	}

	/**
	 * 从channel中移除多个listener
	 * 
	 * @param channel
	 * @param listenerClassNames
	 *            监听者class全名
	 */
	public synchronized void removeListener(String channel, String... listenerClassNames) {
		Set<MessageListener> listeners = listenerMap.get(channel);
		Iterator<MessageListener> iters = listeners.iterator();

		while (iters.hasNext()) {
			if (ArrayUtils.contains(listenerClassNames, iters.next().getClass().getName()))
				iters.remove();
		}

		if (Toolkit.isEmpty(listeners)) {
			listenerMap.remove(channel);
			subscribedChannels.remove(channel);
			unsubscribe(channel);
		}
	}

	public synchronized void removeListener(String channel){
		listenerMap.remove(channel);
		subscribedChannels.remove(channel);
		unsubscribe(channel);
	}
	/**
	 * 不再监听所有channel，相当于关闭
	 */
	public synchronized void stop(){
		for(String channel:listenerMap.keySet()){
			removeListener(channel);
		}
	}
	
	public Map<String, Set<MessageListener>> getListenerMap() {
		return listenerMap;
	}

	public void setListenerMap(Map<String, Set<MessageListener>> listenerMap) {
		this.listenerMap = listenerMap;
	}

	public JedisPubSub getPubsubClient() {
		return pubsubClient;
	}

	public void setPubsubClient(JedisPubSub pubsubClient) {
		this.pubsubClient = pubsubClient;
	}

	public Set<String> getSubscribedChannels() {
		return subscribedChannels;
	}

	public void setSubscribedChannels(Set<String> subscribedChannels) {
		this.subscribedChannels = subscribedChannels;
	}

	public boolean isSubscribe() {
		return isSubscribe;
	}

	public void setSubscribe(boolean isSubscribe) {
		this.isSubscribe = isSubscribe;
	}

}
